package com.learn.lb.hbase;

import org.apache.commons.lang.math.NumberUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

/**
 * @author laibo
 * @since 2019/8/19 16:57
 * ===================================
 * <p> hbase demo
 * <p> 熟悉hbase的客户端api，有些连接并没有关闭
 * ===================================
 */
public class HbaseDemo {

    private static Logger logger = LoggerFactory.getLogger(HbaseDemo.class);

//    private static Configuration configuration;

    private static Connection connection;

    /**
     * 创建表管理类
     */
    private static Admin admin;

    /**
     * rowKey自增
     */
    private static AtomicInteger rowKey = new AtomicInteger(1000);

    static {
        try {
            logger.info("-----------【开始连接hbase】-----------");
            //连接集群
            Configuration configuration = HBaseConfiguration.create();
            // 设置连接参数：hbase-site.xml 配置的quorum
            configuration.set("hbase.zookeeper.quorum", "master");
            // 设置连接参数：zookeeper设置的clientPort，可到zoo.cfg查看
            configuration.set("hbase.zookeeper.clientPort", "2181");
            //创建连接
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
            logger.info("-----------【连接hbase成功】-----------");
        } catch (Exception e) {
            logger.error("-----------【连接zookeeper失败，检查配置是否正确】-----------", e);
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws Exception {
        String tableName = "student";
        createTable(tableName, "info");
        Map<String, Map<String, String>> rowData = new HashMap<>(8);
        rowData.put("info", new HashMap<String, String>(2) {{
            put("name", "王五");
            put("age", "22");
        }});
        putData(tableName, rowData);
        scanAllTableData(tableName);
        getDataByRowKey(tableName, "1000");
        getCell(tableName, "info", "name", "1000", Boolean.FALSE);
        getCell(tableName, "info", "name", "1000", Boolean.TRUE);

    }

    /**
     * 新增表
     *
     * @param tableNameStr 表名
     * @param cfList       列簇集合
     * @throws IOException io异常
     */
    public static void createTable(String tableNameStr, String... cfList) throws IOException {
        logger.info("-----------【开始创建表】-----------");
        if (cfList == null || cfList.length == 0) {
            logger.error("请传入至少一个列簇");
            return;
        }
        logger.info("-----------【开始新增表{}】-----------", tableNameStr);
        long start = System.currentTimeMillis();
        //创建表
        TableName hbaseTableName = TableName.valueOf(tableNameStr);
        if (admin.tableExists(hbaseTableName)) {
            logger.error("-----------【表{}已存在，新增表失败】-----------", tableNameStr);
            return;
        }
        //创建表描述
        HTableDescriptor hTableDescriptor = new HTableDescriptor(hbaseTableName);
        //加入列簇
        Stream.of(cfList).forEach(v -> hTableDescriptor.addFamily(new HColumnDescriptor(v)));
        //创建表
        admin.createTable(hTableDescriptor);
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }

    /**
     * 插入数据
     * 数据结构： key：cf, value：key:columnName, value: columnValue <br/>
     * 其实 hbase 结构就像一个map在嵌套一个map <br/>
     *
     * @param rowData 数据
     */
    public static void putData(String tableNameStr, Map<String, Map<String, String>> rowData) throws Exception {
        TableName hbTableName = createHbTableName(tableNameStr);
        if (null == rowData || rowData.size() == 0) {
            logger.error("请至少添加一行数据");
            return;
        }
        logger.info("-----------【往表{}新增数据】-----------", tableNameStr);
        long start = System.currentTimeMillis();
        // 需要插入数据库的数据集合
        List<Put> putList = new ArrayList<>();
        rowData.forEach((k, v) -> {
            //设置rowKey
            Put rowPut = new Put(String.valueOf(rowKey.getAndAdd(1)).getBytes());
            v.forEach((colName, colValue) -> rowPut.addColumn(Bytes.toBytes(k), Bytes.toBytes(colName), Bytes.toBytes(colValue)));
            putList.add(rowPut);
        });
        //批量插入
        connection.getTable(hbTableName).put(putList);
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }

    /**
     * 查询表的所有数据
     *
     * @param tableNameStr 表名
     */
    public static void scanAllTableData(String tableNameStr) throws Exception {
        TableName hbTableName = createHbTableName(tableNameStr);
        logger.info("-----------【查询表{}的所有数据】-----------", tableNameStr);
        long start = System.currentTimeMillis();
        Table table = connection.getTable(hbTableName);
        ResultScanner resultScanner = table.getScanner(new Scan());
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                logger.info("rowKey:{}\t{}:{}=>{}",
                        new Object[]{
                                Bytes.toString(result.getRow()),
                                Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()),
                                Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()),
                                Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())
                        });
            }
        }
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }

    /**
     * 获取指定rowKey的数据
     *
     * @param tableNameStr 表名
     * @param rowKey       rowKey
     */
    public static void getDataByRowKey(String tableNameStr, String rowKey) throws Exception {
        TableName hbTableName = createHbTableName(tableNameStr);
        logger.info("-----------【根据rowKey查询表{}的数据，rowKey:{}】-----------", tableNameStr, rowKey);
        long start = System.currentTimeMillis();
        Table table = connection.getTable(hbTableName);
        // 新建一个以rowKey作为查询条件的查询对象
        Get get = new Get(Bytes.toBytes(rowKey));
        // 按rowKey查询到的数据
        Result result = table.get(get);
        for (Cell cell : result.rawCells()) {
            logger.info("rowKey:{}\t{}:{}=>{}",
                    new Object[]{
                            Bytes.toString(result.getRow()),
                            Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()),
                            Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()),
                            Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())
                    });
        }
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }


    /**
     * 获取指定rowKey和列的数据
     *
     * @param tableNameStr 表名
     * @param cf           列簇名
     * @param colName      列名
     * @param rowKey       rowKey
     * @param isLatestData 是否查询最新的数据(true：查询最新的一条数据，false：返回所有版本的数据)
     */
    public static void getCell(String tableNameStr, String cf, String colName, String rowKey, boolean isLatestData) throws Exception {
        TableName hbTableName = createHbTableName(tableNameStr);
        logger.info("-----------【获取表{},rowKey:{},{}:{}的数据】-----------", new Object[]{tableNameStr, rowKey, cf, colName});
        long start = System.currentTimeMillis();
        Table table = connection.getTable(hbTableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.setMaxVersions(isLatestData ? NumberUtils.INTEGER_ONE : Integer.MAX_VALUE);
        // 按rowKey查询到的数据
        Result result = table.get(get);
        List<Cell> cells = result.getColumnCells(cf.getBytes(), colName.getBytes());
        for (Cell cell : cells) {
            logger.info("rowKey:{}\t{}:{}=>{}\ttimestamp:{}",
                    new Object[]{
                            rowKey,
                            cf,
                            colName,
                            Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()),
                            cell.getTimestamp()
                    });
        }
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }

    /**
     * 清空表数据
     *
     * @param tableNameStr 表名
     */
    public static void truncateTable(String tableNameStr) throws Exception {
        TableName hbTableName = createHbTableName(tableNameStr);
        logger.info("-----------【清空表{}的数据】-----------", tableNameStr);
        long start = System.currentTimeMillis();
        admin.disableTable(hbTableName);
        admin.truncateTable(hbTableName, Boolean.TRUE);
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }

    /**
     * 根据rowKey删除数据
     *
     * @param tableNameStr 表名
     * @param rowKey       rowKey
     */
    public static void deleteByRowKey(String tableNameStr, String rowKey) throws Exception {
        TableName hbTableName = createHbTableName(tableNameStr);
        logger.info("-----------【根据rowKey:{}删除表:{}的数据】-----------", rowKey, tableNameStr);
        long start = System.currentTimeMillis();
        Table table = connection.getTable(hbTableName);
        //创建删除对象
        Delete delete = new Delete(rowKey.getBytes());
        table.delete(delete);
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }

    /**
     * 添加列簇
     *
     * @param tableNameStr 表名
     * @param columnFamily 列簇
     */
    public static void addColumnFamily(String tableNameStr, String columnFamily) throws Exception {
        TableName hbTableName = createHbTableName(tableNameStr);
        logger.info("-----------【给表{}添加列簇{}】-----------", tableNameStr, columnFamily);
        long start = System.currentTimeMillis();
        // 创建列族对象
        HColumnDescriptor hcd = new HColumnDescriptor(columnFamily);
        admin.addColumn(hbTableName, hcd);
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }

    /**
     * 删除列簇对象
     *
     * @param tableNameStr 表名
     * @param columnFamily 列簇
     */
    public static void deleteColumnFamily(String tableNameStr, String columnFamily) throws Exception {
        TableName hbTableName = createHbTableName(tableNameStr);
        logger.info("-----------【删除表{}的列簇{}】-----------", tableNameStr, columnFamily);
        long start = System.currentTimeMillis();
        admin.deleteColumn(hbTableName, columnFamily.getBytes());
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }


    /**
     * 删除表
     *
     * @param tableNameStr 表名
     */
    public static void dropTable(String tableNameStr) throws Exception {
        TableName hbTableName = createHbTableName(tableNameStr);
        logger.info("-----------【完全删除表{}】-----------", tableNameStr);
        long start = System.currentTimeMillis();
        //将表置为失效状态
        admin.disableTable(hbTableName);
        //删除表
        admin.deleteTable(hbTableName);
        logger.info("耗时：{}ms", (System.currentTimeMillis() - start));
        logger.info("-----------【操作成功】-----------");
    }

    private static TableName createHbTableName(String tableNameStr) throws Exception {
        TableName hbTableName = TableName.valueOf(tableNameStr);
        if (!admin.tableExists(hbTableName)) {
            logger.error("{}表不存在！操作失败", tableNameStr);
            throw new RuntimeException("表不存在，操作失败");
        }
        return hbTableName;
    }


}
